package com.xujian.cdc.sink.doris;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.UUID;

/**
 * @author star xu
 * @date 2023/3/31 10:58
 * @Description:
 * @Slogan: 致敬大师,致敬未来的自己
 */
public class Mysql2DorisWithFlinkSql {

    public static void main(String[] args) throws Exception {
        final ParameterTool params = ParameterTool.fromArgs(args);
        Configuration configuration = new Configuration();
        configuration.setString(RestOptions.BIND_PORT, "8081");
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        // source only supports parallelism of 1
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        final StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        // register a table in the catalog
        tEnv.executeSql(
                "CREATE TABLE table_hash (\n" +
                        " k1 STRING," +
                        " k2 STRING," +
                        " v1 STRING," +
                        " v2 STRING," +
                        "  PRIMARY KEY (k1) NOT ENFORCED" +
                        ") WITH (\n" +
                        "  'connector' = 'mysql-cdc',\n" +
                        "  'hostname' = '127.0.0.1',\n" +
                        "  'port' = '3306',\n" +
                        "  'username' = 'root',\n" +
                        "  'password' = 'qwer1234',\n" +
                        "  'database-name' = 'flink-test',\n" +
                        "  'table-name' = 'table_hash',\n" +
                        "  'server-id' = '36881',\n" +
                        "  'server-time-zone' = 'Asia/Shanghai'\n" +
                        ")");
        //doris table
        tEnv.executeSql(
                "CREATE TABLE table_hash_sink (" +
                        "k1 STRING," +
                        "k2 STRING," +
                        "v1 STRING," +
                        "v2 STRING" +
                        ") " +
                        "WITH (\n" +
                        "  'connector' = 'doris',\n" +
                        "  'fenodes' = '192.168.16.112:8030',\n" +
                        "  'table.identifier' = 'test.table_hash3',\n" +
                        "  'username' = 'test',\n" +
                        "  'password' = 'test@2023',\n" +
                        "  'sink.label-prefix' = '" + UUID.randomUUID() + "'\n" +
                        ")");
        // define a dynamic aggregating query
       /* final Table result = tEnv.sqlQuery("SELECT k1,k2,v1,v2 FROM table_hash");
        result.execute().print();*/

        //insert into mysql table to doris table
        tEnv.executeSql("INSERT INTO table_hash_sink(k1,k2,v1,v2) select k1,k2,v1,v2 from table_hash");
    }
}